package com.los.rabbitmq.confirm;

import com.los.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;

/**
 * @program: rabbitmq
 * @description: confirm模式--异步
 * @author: Yujuan
 * @create: 2018-08-02 16:55
 **/
public class ConfirmSend3 {
    private static final String QUEUE_NAME="los_confirm_queue_3";
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        //获取一个连接
        Connection connection= ConnectionUtils.getConnection();
        //创建一个通道
        Channel channel=connection.createChannel();
        //通过通道创建一个队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //生产者调用confirmSelect，将channel设置为confirm模式
        channel.confirmSelect();
        //未确认的信息标识
        final SortedSet<Long> confirmSet= Collections.synchronizedSortedSet(new TreeSet<>());

        //加一个监听通道
        channel.addConfirmListener(new ConfirmListener() {
            //成功
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if (multiple){
                    System.out.println("----handleAck---multiple:  "+multiple);
                    confirmSet.headSet(deliveryTag+1).clear();
                }else {
                    System.out.println("----handleAck---multiple:  "+multiple);
                    confirmSet.remove(deliveryTag);
                }
            }

            //失败
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if (multiple){
                    System.out.println("----handleNack---multiple:  "+multiple);
                    confirmSet.headSet(deliveryTag+1).clear();
                }else {
                    System.out.println("----handleNack---multiple:  "+multiple);
                    confirmSet.remove(deliveryTag);
                }
            }
        });

        String msg="Hello Confirm sync";

        while (true){
            long seqNo=channel.getNextPublishSeqNo();
            channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
            confirmSet.add(seqNo);
        }

    }
}
